package com.shelpe.services.lexicon.service;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.Predicate;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.metrics.CounterService;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import com.shelpe.services.lexicon.constants.Device;
import com.shelpe.services.lexicon.constants.Location;
import com.shelpe.services.lexicon.entity.AdgroupRecord;
import com.shelpe.services.lexicon.entity.KeywordMetric;
import com.shelpe.services.lexicon.entity.KeywordRecommend;
import com.shelpe.services.lexicon.entity.Scope;
import com.shelpe.services.lexicon.entity.WordInsightMetric;
import com.shelpe.services.lexicon.entity.metrics.InsightMetrics;
import com.shelpe.services.lexicon.repository.mongo.AdgroupRecordRepository;
import com.shelpe.services.lexicon.repository.mongo.KeywordMetricRepository;
import com.shelpe.services.lexicon.repository.mongo.WordInsightMetricRepository;
import com.shelpe.services.lexicon.simba.InsightWordData;
import com.shelpe.services.lexicon.simba.WordClient;
import com.shelpe.services.lexicon.simba.WordClient.CategoryWordResponse;
import com.shelpe.services.lexicon.simba.WordClient.GlobalWordRespone;
import com.shelpe.services.lexicon.simba.WordClient.NetworkWordResponse;
import com.shelpe.services.lexicon.utils.MetricUtils;

import lombok.extern.slf4j.Slf4j;
import rx.Observable;
import rx.functions.Action1;

@Service
@Slf4j
public class InsightMetricService{

	private final static String[] TEMPLATE = new String[0];

	@Autowired
	private WordClient wordClient;
	@Autowired
	private KeywordMetricRepository keywordMetricRepository;
	@Autowired
	private AdgroupRecordRepository adgroupRecordRepository;
	@Autowired
	private WordInsightMetricRepository wordInsightMetricRepository;
	@Value("${simbaproxy.insight.count.perQuery:100}")
	private int countPerQuery;
	@Autowired
	private MongoTemplate mongoTemplate;
	@Autowired
	private CounterService counterService;
	
	public long countByAdgroupRecordIDAndGlobalAndScopeEndTime(String recordID, boolean global, Date endTime){
		Criteria criteria = Criteria.where("adgroupRecordIDs").all(recordID).and("global").is(true);
		criteria = criteria.and("scope.endTime").is(endTime);
		return this.mongoTemplate.count(Query.query(criteria), WordInsightMetric.class);
	}
	
	public List<WordInsightMetric> findByAdgroupIDAndCategoryID(long adgroupID, long categoryID, Date... endTimes){
		long start = System.currentTimeMillis();
		List<Date> dates = new ArrayList<Date>();
		if(endTimes != null){
			for(Date date: endTimes){
				dates.add(date);
			}
		}
		
		Criteria criteria = Criteria.where("adgroupIDs").all(adgroupID).and("categoryID").is(categoryID+"");
		if(dates.size() > 0){
			criteria = criteria.and("scope.endTime").in(dates);
		}
		List<WordInsightMetric> metrics = this.mongoTemplate.find(Query.query(criteria), WordInsightMetric.class);
		log.info("findByAdgroupIDAndCategoryID cost {}ms: {}", (System.currentTimeMillis() - start), adgroupID);
		return metrics;
	}
	
	public List<WordInsightMetric> findByAdgroupRecordID(String adgroupRecordID, Date... endTimes){
		long start = System.currentTimeMillis();
		List<Date> dates = new ArrayList<Date>();
		if(endTimes != null){
			for(Date date: endTimes){
				dates.add(date);
			}
		}
		
		Criteria criteria = Criteria.where("adgroupRecordIDs").all(adgroupRecordID);
		if(dates.size() > 0){
			criteria = criteria.and("scope.endTime").in(dates);
		}
		List<WordInsightMetric> metrics = this.mongoTemplate.find(Query.query(criteria), WordInsightMetric.class);
		log.info("findByAdgroupRecordID cost {}ms: {}", (System.currentTimeMillis() - start), adgroupRecordID);
		return metrics;
	}
	
	public List<WordInsightMetric> findByAdgroupID(long adgroupID, Date... endTimes){
		long start = System.currentTimeMillis();
		List<Date> dates = new ArrayList<Date>();
		if(endTimes != null){
			for(Date date: endTimes){
				dates.add(date);
			}
		}
		
		Criteria criteria = Criteria.where("adgroupIDs").all(adgroupID);
		if(dates.size() > 0){
			criteria = criteria.and("scope.endTime").in(dates);
		}
		List<WordInsightMetric> metrics = this.mongoTemplate.find(Query.query(criteria), WordInsightMetric.class);
		log.info("findByAdgroupID cost {}ms: {}", (System.currentTimeMillis() - start), adgroupID);
		return metrics;
	}
	
	public void syncGlobalInsightData4Recommends(final List<KeywordRecommend> recommends, final Date start, final Date end, 
			final AdgroupRecord record, final Callback completeCallback) {
		final WordChunk chunk = new WordChunk(this.countPerQuery);
		for(KeywordRecommend recommend: recommends){
			chunk.addWord(recommend.getWord());
		}
		log.info("start to sync global insight data: {}", record);
		final WordChunk.ChunkCallback callback = new WordChunk.ChunkCallback(){
			@Override
			public void call(WordChunk chunk, List<KeywordRecommend> krs, WordChunk.ChunkCallback next) {
				if(chunk == null){
					log.info("finish to sync global insight data: {}", record);
					completeCallback.call(recommends);
					return;
				}
				syncGlobalPerChunk(recommends, start, end, record, chunk, next);
			}
		};
		this.syncGlobalPerChunk(recommends, start, end, record, chunk, callback);
	}
	
	public void syncInsightData4Recommends(List<KeywordRecommend> krs, final Date start, final Date end, 
			final AdgroupRecord record, final Callback completeCallback) {
		Scope scope = new Scope();
		scope.setStartTime(start);
		scope.setEndTime(end);
		final List<KeywordRecommend> recommends = this.filterExistings(krs, record, scope);
		
		final WordChunk chunk = new WordChunk(this.countPerQuery);
		for(KeywordRecommend recommend: recommends){
			chunk.addWord(recommend.getWord());
		}
		log.info("start to sync global & category & network insight data: {}", record);
		final WordChunk.ChunkCallback callback = new WordChunk.ChunkCallback(){
			@Override
			public void call(WordChunk chunk, List<KeywordRecommend> krs, WordChunk.ChunkCallback next) {
				if(chunk == null){
					log.info("finish to sync global & category & network insight data: {}", record);
					completeCallback.call(recommends);
					return;
				}
				syncInsightDataPerChunk(recommends, start, end, record, chunk, next);
			}
		};
		this.syncInsightDataPerChunk(recommends, start, end, record, chunk, callback);
	}
	
	public void syncGlobalAndCategoryInsightData4Recommends(final List<KeywordRecommend> recommends, final Date start, final Date end, 
			final AdgroupRecord record, final Callback completeCallback) {
		final WordChunk chunk = new WordChunk(this.countPerQuery);
		for(KeywordRecommend recommend: recommends){
			chunk.addWord(recommend.getWord());
		}
		
		log.info("start to sync global & category insight data: {}", record);
		final WordChunk.ChunkCallback callback = new WordChunk.ChunkCallback(){
			@Override
			public void call(WordChunk chunk, List<KeywordRecommend> krs, WordChunk.ChunkCallback next) {
				if(chunk == null){
					log.info("finish to sync global & category insight data: {}", record);
					completeCallback.call(recommends);
					return;
				}
				syncGlobalAndCategoryInsightDataPerChunk(recommends, start, end, record, chunk, next);
			}
		};
		this.syncGlobalAndCategoryInsightDataPerChunk(recommends, start, end, record, chunk, callback);
	}
	
	public void syncInsightDataPerChunk(final List<KeywordRecommend> recommends, final Date start, final Date end, 
			final AdgroupRecord record, final WordChunk chunk, final WordChunk.ChunkCallback next) {		
		if(chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(chunk.getNext(), recommends, next);
			return;
		}
		log.info("start to sync global & category & network insight data per chunk: {}", record);
		final int totalRuns = 3;
		final AtomicInteger indexer = new AtomicInteger();
		this.syncGlobalData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
		this.syncCategoryData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
		syncNetworkData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
	}
	
	public void syncGlobalAndCategoryInsightDataPerChunk(final List<KeywordRecommend> recommends, final Date start, final Date end, 
			final AdgroupRecord record, final WordChunk chunk, final WordChunk.ChunkCallback next) {	
		if(chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(chunk.getNext(), recommends, next);
			return;
		}
		log.info("start to sync global & category insight data per chunk: {}", record);
		final int totalRuns = 2;
		final AtomicInteger indexer = new AtomicInteger();
		this.syncGlobalData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
		this.syncCategoryData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
	}
	
	public void syncGlobalPerChunk(final List<KeywordRecommend> recommends, final Date start, final Date end, 
			final AdgroupRecord record, final WordChunk chunk, final WordChunk.ChunkCallback next) {	
		if(chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(chunk.getNext(), recommends, next);
			return;
		}
		log.info("start to sync global insight data per chunk: {}", record);
		final int totalRuns = 1;
		final AtomicInteger indexer = new AtomicInteger();
		this.syncGlobalData(chunk, start, end, record, recommends, new Callback(){
			@Override
			public void call(List<KeywordRecommend> krs) {
				if(indexer.incrementAndGet() == totalRuns){
					next.call(chunk.getNext(), recommends, next);
				}
			}
		});
	}

	public void syncGlobalData(final WordChunk chunk, final Date start, final Date end, 
			final AdgroupRecord record, final List<KeywordRecommend> recommends, final Callback next) {
		if(chunk == null || chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(new ArrayList<KeywordRecommend>());
			return;
		}
		
		Observable<GlobalWordRespone> globalObserver = this.wordClient.findGlobalData(chunk.getWords().toArray(TEMPLATE),
				DateFormatUtils.format(start, "yyyy-MM-dd"), DateFormatUtils.format(end, "yyyy-MM-dd"),
				record.getTanentID());
		counterService.increment("meter.shelpe.lexicon.insight.global");
		globalObserver.subscribe(new Action1<GlobalWordRespone>(){
			@Override
			public void call(GlobalWordRespone res) {
				if (res != null && res.getRes()!=null && res.getRes().getDataList() != null && res.getRes().getDataList().getWords() != null) {
					log.debug("globalObserver get success {}: {}", record, res);
					List<WordInsightMetric> metrics = convert2Metric(res.getRes().getDataList().getWords(), start, end, record);
					
					for(KeywordRecommend recommend: recommends){
						InsightMetrics metric = MetricUtils.findWordInsightMetric(metrics, recommend);
						if(metric != null){
							recommend.setInsight(metric);
						}
					}
					counterService.increment("meter.shelpe.lexicon.insight.global.success");
				}else{
					log.error("globalObserver throw failure {}: {}", record, res);
					if(res != null){
						if(res.getError() != null && res.getError().getSubCode() != null){
							counterService.increment("meter.shelpe.lexicon.insight.global.failure." + res.getError().getSubCode());
						}else{
							counterService.increment("meter.shelpe.lexicon.insight.global.failure.unknownError");
						}
					}else{
						counterService.increment("meter.shelpe.lexicon.insight.global.failure.nullRes");
					}
				}
				next.call(recommends);
			}
			
		}, new Action1<Throwable>(){
			@Override
			public void call(Throwable t) {
				log.debug("globalObserver throw exception {}: {}", record, t);
				counterService.increment("meter.shelpe.lexicon.insight.global.throw");
				next.call(new ArrayList<KeywordRecommend>());
			}
		});
	}
	
	public void syncCategoryData(final WordChunk chunk, final Date start, final Date end, 
			final AdgroupRecord record, final List<KeywordRecommend> recommends, final Callback next) {
		if(chunk == null || chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(new ArrayList<KeywordRecommend>());
			return;
		}
		
		Observable<CategoryWordResponse> catObserver = this.wordClient.findCategoryData(chunk.getWords().toArray(TEMPLATE),
				DateFormatUtils.format(start, "yyyy-MM-dd"), DateFormatUtils.format(end, "yyyy-MM-dd"),
				record.getRootCategoryID()+ "", record.getTanentID());
		counterService.increment("meter.shelpe.lexicon.insight.network");
		catObserver.subscribe(new Action1<CategoryWordResponse>(){
			@Override
			public void call(CategoryWordResponse res) {
				if (res != null && res.getRes()!=null && res.getRes().getDataList() != null && res.getRes().getDataList().getWords() != null) {
					log.debug("catObserver get success {}: {}", record, res);
					convert2Metric(res.getRes().getDataList().getWords(), start, end, record);
					counterService.increment("meter.shelpe.lexicon.insight.category.success");
				}else{
					log.error("catObserver throw failure {}: {}", record, res);
					if(res != null){
						if(res.getError() != null && res.getError().getSubCode() != null){
							counterService.increment("meter.shelpe.lexicon.insight.category.failure." + res.getError().getSubCode());
						}else{
							counterService.increment("meter.shelpe.lexicon.insight.category.failure.unknownError");
						}
					}else{
						counterService.increment("meter.shelpe.lexicon.insight.category.failure.nullRes");
					}
				}
				next.call(recommends);
			}
			
		}, new Action1<Throwable>(){
			@Override
			public void call(Throwable t) {
				log.debug("catObserver throw exception {}: {}", record, t);
				counterService.increment("meter.shelpe.lexicon.insight.category.throw");
				next.call(new ArrayList<KeywordRecommend>());
			}
			
		});
	}
	
	public void syncNetworkData(final WordChunk chunk, final Date start, final Date end, 
			final AdgroupRecord record, final List<KeywordRecommend> recommends, final Callback next) {
		if(chunk == null || chunk.getWords() == null || chunk.getWords().size() == 0){
			next.call(new ArrayList<KeywordRecommend>());
			return;
		}
		
		Observable<NetworkWordResponse> networkObserver = this.wordClient.findNetworkData(chunk.getWords().toArray(TEMPLATE),
				DateFormatUtils.format(start, "yyyy-MM-dd"), DateFormatUtils.format(end, "yyyy-MM-dd"),
				record.getTanentID());
		counterService.increment("meter.shelpe.lexicon.insight.network");
		networkObserver.subscribe(new Action1<NetworkWordResponse>(){
			@Override
			public void call(NetworkWordResponse res) {
				if (res != null && res.getRes()!=null && res.getRes().getDataList() != null && res.getRes().getDataList().getWords() != null) {
					log.debug("networkObserver get success {}: {}", record, res);
					List<WordInsightMetric> metrics = convert2Metric(res.getRes().getDataList().getWords(), start, end, record);
					CollectionUtils.filter(metrics, new Predicate(){
						@Override
						public boolean evaluate(Object object) {
							WordInsightMetric metric = (WordInsightMetric) object;
							return metric.getDevice() == Device.mobile;
						}
					});
					for(KeywordRecommend recommend: recommends){
						InsightMetrics metric = MetricUtils.findWordInsightMetric(metrics, recommend);
						if(metric != null){
							recommend.setMobileInsight(metric);
						}
					}
					counterService.increment("meter.shelpe.lexicon.insight.network.success");
				}else{
					log.error("networkObserver throw failure {}: {}", record, res);
					if(res != null){
						if(res.getError() != null && res.getError().getSubCode() != null){
							counterService.increment("meter.shelpe.lexicon.insight.network.failure." + res.getError().getSubCode());
						}else{
							counterService.increment("meter.shelpe.lexicon.insight.network.failure.unknownError");
						}
					}else{
						counterService.increment("meter.shelpe.lexicon.insight.network.failure.nullRes");
					}
				}
				next.call(recommends);
			}
			
		}, new Action1<Throwable>(){
			@Override
			public void call(Throwable t) {
				log.debug("networkObserver throw exception {}: {}", record, t);
				counterService.increment("meter.shelpe.lexicon.insight.network.throw");
				next.call(new ArrayList<KeywordRecommend>());
			}
			
		});
	}
	
	private List<KeywordRecommend> filterExistings(List<KeywordRecommend> recommends, AdgroupRecord record, Scope scope){
		long startTS = System.currentTimeMillis();
		Map<String, KeywordRecommend> remainings = new HashMap<String, KeywordRecommend>();
		
		List<String> globalMetricIDs = new ArrayList<String>();
		List<String> mobileMetricIDs = new ArrayList<String>();
		for(KeywordRecommend recommend: recommends){
			globalMetricIDs.add(WordInsightMetric.calcMetricID(recommend.getWord(), scope, "g"));
			mobileMetricIDs.add(WordInsightMetric.calcMetricID(recommend.getWord(), scope, "d[" + Device.mobile.name() + "]l[" + Location.all.name() + "]"));
		}
		
		List<WordInsightMetric> metrics = this.wordInsightMetricRepository.findByMetricIDIn(globalMetricIDs);
		for(KeywordRecommend recommend: recommends){
			InsightMetrics metric = MetricUtils.findWordInsightMetric(metrics, recommend);
			if(metric != null){
				recommend.setInsight(metric);
			}else{
				remainings.put(recommend.getWord(), recommend);
			}
		}
		
		metrics = this.wordInsightMetricRepository.findByMetricIDIn(mobileMetricIDs);
		for(KeywordRecommend recommend: recommends){
			InsightMetrics metric = MetricUtils.findWordInsightMetric(metrics, recommend);
			if(metric != null){
				recommend.setMobileInsight(metric);
			}else{
				remainings.put(recommend.getWord(), recommend);
			}
		}
		
		log.info("still need sync {} in {} recommends: {}", remainings.size(), recommends.size(), record);
		recommends = new ArrayList<KeywordRecommend>();
		for(Map.Entry<String, KeywordRecommend> entry: remainings.entrySet()){
			recommends.add(entry.getValue());
		}
		log.info("filter existings costs {}ms: {}", System.currentTimeMillis() - startTS, record);
		return recommends;
	}
	
	
	public void merge2WordInsightMetric(AdgroupRecord record){
		List<KeywordMetric> kms = this.keywordMetricRepository.findByAdgroupRecordID(record.getRecordID());
		
		List<WordInsightMetric> metrics = new ArrayList<WordInsightMetric>(kms.size() * 2);
		for(KeywordMetric km: kms){
			WordInsightMetric metric = new WordInsightMetric();
			BeanUtils.copyProperties(km, metric);
			metric.setWord(km.getWord());
			metric.setDevice(km.getDevice());
			metric.setScope(km.getScope());
			metric.setCategoryID(km.getCategoryID());
			metric.setCity(km.getCity());
			metric.setGlobal(metric.isGlobal());
			metric.setLocation(km.getLocation());
			metric.setProvince(metric.getProvince());
			metric.setMetric(km.getMetric());
			metric.setupMetricID();
			
			WordInsightMetric existing = this.wordInsightMetricRepository.findOne(km.getMetricID());
			if(existing != null){
				log.debug("the metric {} is exist", km.getMetricID());
				metric = existing;
				metric.setLastModifiedTime(new Date());
			}else{
				metric.setCreateTime(new Date());
				metric.setLastModifiedTime(new Date());
			}
			metric.addAdgroupID(record.getAdgroupID());
			metric.addAdgroupRecordID(record.getRecordID());
			metrics.add(metric);
		}
		this.wordInsightMetricRepository.save(metrics);
		
		record.setMerged(true);
		record.setLastModifiedTime(new Date());
		this.adgroupRecordRepository.save(record);
	}
	
	private List<WordInsightMetric> convert2Metric(List<InsightWordData> words, Date start, Date end, AdgroupRecord record){
		long startTS = System.currentTimeMillis();
		List<WordInsightMetric> metrics = new ArrayList<WordInsightMetric>();
		Scope scope = new Scope();
		scope.setEndTime(end);
		scope.setStartTime(start);

		for (InsightWordData data : words) {
			WordInsightMetric km = new WordInsightMetric();
			km.from(data);
			km.setScope(scope);
			km.setupMetricID();
			
			WordInsightMetric existing = this.wordInsightMetricRepository.findOne(km.getMetricID());
			if(existing != null){
				log.debug("the metric {} is exist", km.getMetricID());
				km = existing;
				km.setLastModifiedTime(new Date());
			}else{
				km.setCreateTime(new Date());
				km.setLastModifiedTime(new Date());
			}
			km.addAdgroupID(record.getAdgroupID());
			km.addAdgroupRecordID(record.getRecordID());
			
			metrics.add(km);
			record.addKeywordMetric(km.getMetric());
		}
		this.wordInsightMetricRepository.save(metrics);
		
		record.setLastModifiedTime(new Date());
		this.adgroupRecordRepository.save(record);
		
		log.info("store to db cost {}ms: {}", (System.currentTimeMillis() - startTS), record);
		return metrics;
	}
	
	public static interface Callback{
		void call(List<KeywordRecommend> recommends);
	}
}
